package pip

import (
	"fmt"
	"gitee.com/ChanYeung/go-utils/paralla"
	"log"
	"sync"
	"time"
)

// RPipeHandler Handler接口
type RPipeHandler interface {
	Name() string
	PipeProcess(ctx *RPipeContext)
}

// RPipeContext Pipeline中传递的Context
type RPipeContext struct {
	Ctx    *SimpleContext
	Err    error // 只填充阻碍全流程的错误
	IsStop bool  // 是否停止整个流程，若为true，跳过后续pipeline
	//添加其他自定义实现的上下文参数
}

// RPipeLine Pipeline
type RPipeLine struct {
	Handlers        []RPipeHandler // 按添加顺序执行
	CurrentPipeline []*RPipeLine   // 并发pipeline使用
	Ctx             *RPipeContext
	PipeName        string
	Logger          *log.Logger //日志logger可以 自定义包装下，实现更复杂的日志信息记录
}

func NewRPipeLine(pipCtx *RPipeContext, name string) *RPipeLine {
	return &RPipeLine{
		Handlers: make([]RPipeHandler, 0),
		Ctx:      pipCtx,
		PipeName: name,
		Logger:   log.Default(),
	}
}

func (pipe *RPipeLine) AddPipe(handler RPipeHandler) *RPipeLine {
	pipe.Handlers = append(pipe.Handlers, handler)
	return pipe
}

func (pipe *RPipeLine) Process() error {

	pipe.Logger.Println("handlerName_step_cost_startTime")
	for i, handler := range pipe.Handlers {
		if pipe.Ctx.IsStop {
			pInfo := fmt.Sprintf("%v_%v", handler.Name(), i)
			pipe.Logger.Println("skip pipe", pInfo)
			continue
		}
		startPipe := time.Now().UnixNano()
		handler.PipeProcess(pipe.Ctx)
		processPipeCost := time.Now().UnixNano() - startPipe
		pipeInfo := fmt.Sprintf("%v_%v_%v_%v", handler.Name(), i, processPipeCost/1000000, startPipe)
		pipe.Logger.Println(pipe.PipeName, pipeInfo)
	}
	return pipe.Ctx.Err
}

func (pipe *RPipeLine) AddCurrentPipeline(pipeline *RPipeLine) *RPipeLine {
	pipe.CurrentPipeline = append(pipe.CurrentPipeline, pipeline)
	return pipe
}

func (pipe *RPipeLine) ProcessCurrentPipeline() error {
	start := time.Now().UnixNano()
	var wg sync.WaitGroup
	wg.Add(len(pipe.CurrentPipeline))

	for i := range pipe.CurrentPipeline {
		go func(i int) {
			defer wg.Done()
			p := pipe.CurrentPipeline[i]
			err := paralla.PanicRecoverDecorator(func() error {
				log.Printf("RPipeline_Process||CurrentPipeline action=【%v】start index{%v}",
					p.PipeName, i)
				startPipe := time.Now().UnixNano()

				p.Process()

				processPipeCost := time.Now().UnixNano() - startPipe
				log.Printf("RPipeline_Process||CurrentPipeline action=【%v】finish index{%v} , cost %v",
					p.PipeName, i, processPipeCost/1000000)

				return p.Ctx.Err
			}, pipe.Ctx.Ctx)

			if err != nil {
				log.Printf("RPipeline_Process||CurrentPipeline action=【%v】process error index{%v}||err=%v",
					p.PipeName, i, err)
			}
		}(i)

	}

	wg.Wait()
	processPipeCost := time.Now().UnixNano() - start
	log.Printf("RPipeline_Process||CurrentPipeline【%v】all finished , cost %v",
		pipe.PipeName, processPipeCost/1000000)

	return pipe.Ctx.Err
}
